/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package thesis.ramstorage.application;

import java.math.BigInteger;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Observable;
import java.util.Random;
import java.util.SortedMap;
import java.util.TreeMap;

import thesis.ramstorage.core.Fallible;
import thesis.ramstorage.core.PhysicalNode;
import thesis.ramstorage.core.VirtualNode;
import thesis.ramstorage.math.Evaluator;
import thesis.ramstorage.math.HashFunction;
import thesis.ramstorage.math.StatisticParameters;
import thesis.ramstorage.util.EvaluatedParameters;
import thesis.ramstorage.util.NodeParameters;

/**
 *
 * @author hlminh
 */
public class NetworkControl extends Observable {

    //suppose we have a network control that holds on node records
    //this is unrealistic, only for simulation
    //next version, we must implement a lookup schema, so that this network control
    //holds some node records
    SystemConfiguration config;
    private HashFunction hashFunction = null;//
    private final SortedMap<BigInteger, VirtualNode> circle = new TreeMap<BigInteger, VirtualNode>();
    private BigInteger currentPointer = new BigInteger("0");
    public static List<PhysicalNode> nodes = null;
    private long averageLoad; //average load per virtual node
    private long minLoad;
    private int networkStatus;
//	private long maxHashValue;

//	public static long[] allNodeLoad;
    public NetworkControl(SystemConfiguration config) {
        // TODO Auto-generated constructor stub
        this.config = config;
        nodes = new ArrayList<PhysicalNode>(config.getNumberOfPhysicalNode());
        this.hashFunction = new HashFunction(config.getHashKeySize());
//		allNodeLoad = new long[nodes.length];
//		this.maxHashValue = (long)Math.pow(2, config.getHashKeySize());
    }

    public void initialize() {
        List<VirtualNode> listVNode = new ArrayList<VirtualNode>();
        for (int i = 0; i < config.getNumberOfPhysicalNode(); i++) {
            PhysicalNode physicalNode = new PhysicalNode();
            physicalNode.setNumberOfVirtualNode(randomNodeCapacity());
            physicalNode.setID(i);
            physicalNode.startUp();
            nodes.add(physicalNode);
            listVNode.addAll(constructVirtualNode(physicalNode));
        }
        //this makes random distribution of all node on circle
        Collections.shuffle(listVNode, new Random());
        dispatchVirtualNode(listVNode);
//		init initial data
        for (int i = 0; i < config.getInitialData(); i++) {
            DataObj dobj = new DataObj();
            dobj.setId(i);
            try {
                this.put(dobj);
            } catch (Exception e) {
                // TODO Auto-generated catch block
//				e.printStackTrace();
                System.out.println(e.getMessage());
            }
        }
    }

    private List<VirtualNode> constructVirtualNode(PhysicalNode physicalNode) {
        // TODO Auto-generated method stub
        List<VirtualNode> result = new ArrayList<VirtualNode>();
        for (int i = 0; i < physicalNode.getNumberOfVirtualNode(); i++) {
            VirtualNode vNode = new VirtualNode();
            vNode.setIndex(i);
            vNode.setpNode(physicalNode);
            result.add(vNode);
        }
        return result;
    }

    private void checkNoVNodeInContinuousPositions(List<VirtualNode> listVNode) {
        for (int i = 0; i < listVNode.size(); i++) {
            System.out.print(" " + listVNode.get(i).getpNode().getID());
        }
        for (int i = 0; i < listVNode.size(); i++) {
            VirtualNode curNode = listVNode.get(i);
            VirtualNode preNode = null;
            VirtualNode nextNode = null;
            if (i >= 1) {
                preNode = listVNode.get(i - 1);
            }
            if (i + 1 == listVNode.size()) {
                nextNode = new VirtualNode();
            } else {
                nextNode = listVNode.get(i + 1);
            }

            //System.out.println("aa " + curNode.getPhysicalNodeId());
            if (preNode != null && curNode.getpNode().getID() == preNode.getpNode().getID()) {
                //found
                int j = i + 1;
                while (nextNode.getpNode().getID() ==
                        curNode.getpNode().getID() && j < listVNode.size()) {
                    nextNode = listVNode.get(j);
                    j++;
                }
                //swap nextNode with currNode
                //listVNode.remove(nextNode);
                listVNode.add(i, nextNode);
                listVNode.remove(j + 1);
                //listVNode.get(j) = curNode;
                }

        }

    }

    private void dispatchVirtualNode(List<VirtualNode> listVNode) {

        //random
        //
        checkNoVNodeInContinuousPositions(listVNode);
        for (int i = 0; i < listVNode.size(); i++) {
            System.out.print(" " + listVNode.get(i).getpNode().getID());
        }
        if ("RANDOM".equals(config.getProtocol())) {
            //			BigInteger hashKey = hashFunction.generateSHA1(String.valueOf(listVNode.getID()));
            //			circle.put(hashKey, listVNode);
            } //			System.out.println(" successfully add virtual node " + node.getID() + " at " + hashKey.longValue());
        //		circle.
        else {
            long keyspace = (long) Math.pow(2, config.getHashKeySize());
            int nodeSize = listVNode.size();
            averageLoad = (long) (keyspace / nodeSize);
            long assignedKeys = 0;
            //			double rate = 1D;

            for (VirtualNode vNode : listVNode) {
                BigInteger key = new BigInteger(String.valueOf((long) (assignedKeys)));// + keySizeEachNode * rate)));
                circle.put(key, vNode);
                vNode.setID(key.longValue());
                //update reference to physical node
                vNode.setCapacity(this.averageLoad);
                PhysicalNode pNode = vNode.getpNode();
                pNode.setCapacity(averageLoad * pNode.getNumberOfVirtualNode());
                assignedKeys += averageLoad;
            }
        }
        //must remember successorKey for next join
        //    	 currentPointer = successorKey;

    }

//	private Long generateVirtualID(int index , long physicalID) {
//		// TODO Auto-generated method stub
//		return physicalID * config.getMaxCapacity() + index;
//	}
    public long randomNodeCapacity() {
        // TODO Auto-generated method stub
        return Math.round(((Math.random() * 100)) % (config.getMaxCapacity() - config.getMinCapacity()) + config.getMinCapacity());///from 2 -> 5
    }

    public void simulated() {
        // TODO Auto-generated method stub
    }

    public boolean isManualSimulation() {
        // TODO Auto-generated method stub
        return ("manual".equalsIgnoreCase(config.getSimulatedMode()));
    }

    public String put(DataObj dataObj) throws Exception {
        // TODO Auto-generated method stub
        String res = "OK";
        BigInteger hash = hashFunction.generateSHA1(String.valueOf(dataObj.getId()));
        if (!circle.containsKey(hash)) {
            SortedMap<BigInteger, VirtualNode> tailMap = circle.tailMap(hash);
            hash = tailMap.isEmpty() ? circle.firstKey() : tailMap.firstKey();
        }
        VirtualNode node = circle.get(hash);
        //check if node is shutting down..
        PhysicalNode pNode = node.getpNode();
        synchronized (pNode) {
//            PhysicalNode pNode = NetworkControl.nodes.get(node.getPhysicalNodeId().intValue());
            if (pNode.overload()) {
                //join new node
                long newId = this.nodes.size();
                this.lookupAndJoin(newId);
                pNode.setState(Fallible.OK);
                res = "A new server is joined because of overload";
            }
            Long l = pNode.getLoad();
            if (l == pNode.getCapacity()) {
                throw new Exception("Node of " + pNode.getID() + " is overload. can not input data!");
            } else if (l / pNode.getCapacity() * 100 > SystemConfiguration.overLoadClaim) {
                //            pNode.setState(state);
                pNode.setState(Fallible.HEAVYLOAD);
                res = "heavy load found. please backup data or join new server";
            }
            synchronized (node) {
                if (node.isUp()) {
                    node.put(dataObj.getId(), dataObj);
                } else {
                    //net to implement
                }
            }
        }
//	      this.numberOfObject ++;
        return res;
    }

    public DataObj get(Long id) {
        BigInteger hash = hashFunction.generateSHA1(id.toString());
        synchronized (circle) {
            if (!circle.containsKey(hash)) {
                SortedMap<BigInteger, VirtualNode> tailMap = circle.tailMap(hash);
                hash = tailMap.isEmpty() ? circle.firstKey() : tailMap.firstKey();
            }
            VirtualNode node = circle.get(hash);
            if (!node.isUp()) {
                //find other node
            } else {
                return node.get(id);
            }
            return null;
        }
    }

    public long getNodeLoad(long ID) {
        PhysicalNode pNode = getPhysicalNode(ID);
        synchronized (pNode) {
            if (!pNode.isUp()) {
                return 0;
            }
            return pNode.getLoad();
        }
//		System.out.println(" pnode id " + ID + " has vNode ids " + vNode);
    }

    private PhysicalNode getPhysicalNode(Long id) {
        // TODO Auto-generated method stub
        return nodes.get(id.intValue());
    }

    private VirtualNode getVirtualNode(Long id) {
        // TODO Auto-generated method stub
//		BigInteger hash = hashFunction.generateSHA1(id.toString());
//		if (!circle.containsKey(hash)) {
//			SortedMap<BigInteger, VirtualNode> tailMap = circle.tailMap(hash);
//			hash = tailMap.isEmpty() ? circle.firstKey() : tailMap.firstKey();
//		}
        return circle.get(new BigInteger(id.toString()));
//		return circle.get(new BigInteger(id.toString()));
    }

    public int getSize() {
        // TODO Auto-generated method stub
        return this.nodes.size();
    }

    public long getNumberOfVirtualNode(Integer id) {
        // TODO Auto-generated method stub
        return this.nodes.get(id.intValue()).getNumberOfVirtualNode();
    }

    public EvaluatedParameters evaluate() {
        EvaluatedParameters eva = new EvaluatedParameters();
        for (PhysicalNode n : nodes) {
            if (n.getState() == Fallible.DOWN) {
                continue;
            }
            NodeParameters np = new NodeParameters(getNodeLoad(n.getID()));
//			if(n.getNumberOfVirtualNode() == 0){
//				System.out.println(n.getID() + " ku nay");
//				break;
//			}
            np.setCapacity(n.getNumberOfVirtualNode());
            eva.getNodeParams().put(n.getID(), np);//getNodeLoad(n.getID());
        }
        return eva;
    }

    public void flushData() {
        // TODO Auto-generated method stub
        for (BigInteger k : this.circle.keySet()) {
//            this.circle.get(k).flushData();
        }
    }

    public String join(long pNodeId) throws Exception {
        // TODO Auto-generated method stub
        System.out.println("Node joins follow " + config.getProtocol() + " strategy");
        PhysicalNode newNode = addPhysicalNode(pNodeId);
        if (newNode.getState() == Fallible.OK) {
            return "node is running, can not join again.";
        } else {
            newNode.setState(Fallible.OK);
        }
        BigInteger bi = currentPointer;
        SortedMap<BigInteger, VirtualNode> tailMap = circle.tailMap(bi.add(new BigInteger("1")));
        BigInteger successorKey = tailMap.isEmpty() ? circle.firstKey() : tailMap.firstKey();
        List<VirtualNode> vNodes = constructVirtualNode(newNode);
        newNode.setCapacity(Long.MAX_VALUE);
        for (VirtualNode vNode : vNodes) {
            BigInteger newKey = null;
            if (successorKey.doubleValue() == 0) {
                newKey = bi.add(new BigInteger(String.valueOf((long) Math.pow(2, config.getHashKeySize())))).divide(new BigInteger("2"));
            } else {
                newKey = bi.add(successorKey).divide(new BigInteger("2"));
            }
            this.minLoad = Math.abs(bi.subtract(newKey).longValue());
            vNode.setID(newKey.longValue());
            circle.put(newKey, vNode);
            VirtualNode successorNode = circle.get(successorKey);
            shareData(successorNode, vNode);
            //update to revelant virtual node capacity
            vNode.setCapacity(this.minLoad);
            successorNode.setCapacity(this.minLoad);
            bi = successorKey;//.add(new BigInteger("1"));
            currentPointer = successorKey;
            tailMap = circle.tailMap(bi.add(new BigInteger("1")));
            successorKey = tailMap.isEmpty() ? circle.firstKey() : tailMap.firstKey();
        }
        this.averageLoad = (this.circle.size() - vNodes.size()) * averageLoad / this.circle.size();
        newNode.setCapacity(averageLoad * newNode.getNumberOfVirtualNode());//need to update this average
        //System.out.println("current PT " + currentPointer);
        return "node " + pNodeId + " is joined.";
    }

    //load equally and random join
    public void randomlyJoin(long pNodeId) throws Exception {
        // TODO Auto-generated method stub
        System.out.println("Node joins follow " + config.getProtocol() + " strategy");
        PhysicalNode newNode = addPhysicalNode(pNodeId);

        List<VirtualNode> vNodes = constructVirtualNode(newNode);
        newNode.setCapacity(Long.MAX_VALUE);

        for (VirtualNode vNode : vNodes) {
            //random a key
            String seed = String.valueOf(Math.random());
            BigInteger bi = hashFunction.generateSHA1(seed);
//		   	circle.
            //
            SortedMap<BigInteger, VirtualNode> tailMap = circle.tailMap(bi);//.add(new BigInteger("1")));
            BigInteger successorKey = tailMap.isEmpty() ? circle.firstKey() : tailMap.firstKey();

            SortedMap<BigInteger, VirtualNode> headMap = circle.headMap(bi);//.add(new BigInteger("1")));
            BigInteger preSuccessorKey = headMap.isEmpty() ? circle.lastKey() : headMap.lastKey();

            BigInteger newKey = null;
            if (successorKey.doubleValue() == 0) {
                newKey = preSuccessorKey.add(new BigInteger(String.valueOf((long) Math.pow(2, config.getHashKeySize())))).divide(new BigInteger("2"));
            } else {
                newKey = preSuccessorKey.add(successorKey).divide(new BigInteger("2"));
            }
            //attempt to transfer
            VirtualNode successorNode = circle.get(successorKey);
            long newLoad = Math.abs(bi.subtract(newKey).longValue());

            vNode.setID(newKey.longValue());
            circle.put(newKey, vNode);

            shareData(successorNode, vNode);
            if (this.minLoad > newLoad) {
                minLoad = newLoad;
            }

        }
        this.averageLoad = (this.circle.size() - vNodes.size()) * averageLoad / this.circle.size();
        newNode.setCapacity(averageLoad * newNode.getNumberOfVirtualNode());//need to update this average
    }

    private PhysicalNode addPhysicalNode(long pNodeId) {
        if (nodes.size() <= pNodeId) {
            PhysicalNode node = new PhysicalNode();
            node.setNumberOfVirtualNode(randomNodeCapacity());
            nodes.add(node);
            return node;
        } else {
            PhysicalNode node = nodes.get((int) pNodeId);
            //node.setState(Fallible.OK);
            return node;
        }
    }

    public String previousLocJoin(long pNodeId) throws Exception {
        PhysicalNode newNode = addPhysicalNode(pNodeId);
        if (newNode.getState() == Fallible.OK) {
            return "node is running, can not join again.";
        } else {
            newNode.setState(Fallible.OK);
        }
        for (long key : newNode.getVirtualNodeIds()) {
            VirtualNode vNode = new VirtualNode();
            vNode.setID(key);
            vNode.setpNode(newNode);
            vNode.setFailState(Fallible.OK);
            BigInteger bkey = new BigInteger(String.valueOf(key));
            SortedMap<BigInteger, VirtualNode> tailMap = circle.tailMap(bkey);//.add(new BigInteger("1")));
            BigInteger successorKey = tailMap.isEmpty() ? circle.firstKey() : tailMap.firstKey();
            VirtualNode n = circle.get(successorKey);
            circle.put(bkey, vNode);
//                if(n.getFailState() == Fallible.OK){
//                    assignNodeAndTransferData(startKeySearch, successorKey, vNode);
            getbackData(n, vNode);
//                }
            }
        return "node " + pNodeId + " is joined.";
    }

    public String lookupAndJoin(long pNodeId) throws Exception {
        PhysicalNode newNode = addPhysicalNode(pNodeId);
        newNode.flush();
        if (newNode.getState() == Fallible.OK) {
            return "node is running, can not join again.";
        } else {
            newNode.setState(Fallible.OK);
        }
        BigInteger startKeySearch = circle.firstKey();//node co key min of circle
        List<VirtualNode> vNodes = constructVirtualNode(newNode);
        newNode.setCapacity(Long.MAX_VALUE);
        System.out.println("average load " + this.averageLoad);
        System.out.println("average load " + this.minLoad);
        for (VirtualNode vNode : vNodes) {
            
            vNode.setCapacity(this.averageLoad);
            SortedMap<BigInteger, VirtualNode> tailMap = circle.tailMap(startKeySearch.add(new BigInteger("1")));
            BigInteger successorKey = tailMap.isEmpty() ? circle.firstKey() : tailMap.firstKey();
            int i = circle.size();
            int j = 0;
            boolean noFound = false;
            while (true) {
                if (j == i) {
                    noFound = true;
                    break;//;
                }
                VirtualNode n = circle.get(successorKey);
                if (n.getCapacity() >= 2 * this.minLoad) {
//	        		transfer
                    assignNodeAndTransferData(startKeySearch, successorKey, vNode);
                    break;
                }
                startKeySearch = successorKey;
                successorKey = successorKey.add(new BigInteger("1"));
                tailMap = circle.tailMap(successorKey);
                successorKey = tailMap.isEmpty() ? circle.firstKey() : tailMap.firstKey();
                j++;
            }
            if (noFound) {
                //can not find
                noFound = false;
                j = 0;
                while (true) {
                    if (j == i) {
                        assignNodeAndTransferData(startKeySearch, successorKey, vNode);
                        break;
//						throw new Exception("Can not insert virtual node ID " + vNode.getID());
                    }
                    VirtualNode n = circle.get(successorKey);
                    if (n.getCapacity() > this.minLoad) {
//		        		transfer
                        assignNodeAndTransferData(startKeySearch, successorKey, vNode);
                        break;
                    }
                    startKeySearch = successorKey;
                    successorKey = successorKey.add(new BigInteger("1"));
                    tailMap = circle.tailMap(successorKey);
                    successorKey = tailMap.isEmpty() ? circle.firstKey() : tailMap.firstKey();
                    j++;
                }
            }
        }
        this.averageLoad = (this.circle.size() - vNodes.size()) * averageLoad / this.circle.size();
        newNode.setCapacity(averageLoad * newNode.getNumberOfVirtualNode());//need to update this average
        return "node " + pNodeId + " is joined.";
    }

    private synchronized void assignNodeAndTransferData(BigInteger bi, BigInteger successorKey, VirtualNode target) throws Exception {
        BigInteger newKey = null;
        if (successorKey.doubleValue() == 0) {
            newKey = bi.add(new BigInteger(String.valueOf((long) Math.pow(2, config.getHashKeySize())))).divide(new BigInteger("2"));
        } else {
            newKey = bi.add(successorKey).divide(new BigInteger("2"));
        }
        //		 this.minLoad = Math.abs(successorKey.subtract(newKey).longValue());
        long newLoad = Math.abs(bi.subtract(newKey).longValue());
        VirtualNode successorNode = circle.get(successorKey);
        target.setID(newKey.longValue());
        circle.put(newKey, target);
        shareData(successorNode, target);
        if (this.minLoad > newLoad) {
            minLoad = newLoad;
        }
        target.setCapacity(newLoad);
        successorNode.setCapacity(newLoad);
        bi = bi.add(new BigInteger("1"));
    }

    public Map<Long, VirtualNode> printOut() {
        // TODO Auto-generated method stub
        String po = "";
        Map<Long, VirtualNode> virtualNodes = new TreeMap<Long, VirtualNode>();
        for (BigInteger k : this.circle.keySet()) {
            po += "," + circle.get(k).printOut();
            VirtualNode n = circle.get(k);
            VirtualNode rNode = new VirtualNode();
            rNode.setpNode(n.getpNode());//.getID());
            rNode.setID(n.getID());
            virtualNodes.put(n.getID(), rNode);
        }
        System.out.println("[" + po.substring(1) + "]");
        return virtualNodes;
    }

    public void nodePrint(long nodeId) {
        // TODO Auto-generated method stub
        PhysicalNode node = getPhysicalNode(nodeId);
        if (!node.isUp()) {
            System.out.println("node is shutdown.");
        }
        String o = "";
        for (Long k : node.getVirtualNodeIds()) {
            o += "," + k;
        }
        System.out.println("[" + o.substring(1) + "]");
    }

    public void shutdown(int nodeId) {
        // TODO Auto-generated method stub
        PhysicalNode pNode = getPhysicalNode(Long.valueOf(nodeId));
        if (!pNode.isUp()) {
            return;
        }
        pNode.setState(Fallible.DOWN);
        int size = pNode.getVirtualNodeIds().size();

        minLoad = averageLoad;

        for (Long vID : pNode.getVirtualNodeIds()) {
            VirtualNode vNode = getVirtualNode(vID);
            vNode.setFailState(Fallible.DOWN);
            SortedMap<BigInteger, VirtualNode> tailMap = circle.tailMap(new BigInteger(String.valueOf(vNode.getID() + 1)));
            //check if successorNode is refer same physicalNode
            ///...
            ///end check...
            BigInteger successorKey = tailMap.isEmpty() ? circle.firstKey() : tailMap.firstKey();
            VirtualNode successorNode = circle.get(successorKey);
            transferData(vNode, successorNode);
            //reset capacity
            circle.remove(new BigInteger(String.valueOf(vNode.getID())));
            vNode = null;//prevent memory leak;
        }
//		this.averageLoad = update
        this.averageLoad = (this.circle.size() + size) * averageLoad / this.circle.size();
//		this.nodes.remove(pNode);
        pNode.terminate();
//		pNode= null;
        //use for generate chart
        setChanged();
        StatisticParameters stats = this.statistic();
        notifyObservers(stats);
        //notifyObservers(sp.getStandardDeviation());

    }

    private synchronized void transferData(VirtualNode srcNode, VirtualNode targetNode) {
        List<Long> keys = new ArrayList<Long>(srcNode.listKeys());
        System.out.println(srcNode.getID() + " transfered key count " + keys.size() + " for " + targetNode.getID());
        targetNode.setCapacity(srcNode.getCapacity() + targetNode.getCapacity());
        for (Long storedKey : keys) {
            //transfer to new node
            try {
                targetNode.put(storedKey, srcNode.get(storedKey));
            } catch (Exception e) {
                // TODO Auto-generated catch block
                System.out.println(e.getMessage());
                System.out.println("warning! can lost data.");
                //				e.printStackTrace();
            }
            //				srcNode.remove(storedKey);
        }
    }

    public synchronized long getMinLoad() {
        return this.minLoad;
    }

    public String processKey(String line) {
        // TODO Auto-generated method stub
        String result = "";
        if (line.contains("get")) {
//			Long key = Long.valueOf(line.substring(3).trim());
//			System.out.println("key " + key + " has value " + this.get(Long.valueOf(key)));
        } else if (line.contains("delete")) {
            Long key = Long.valueOf(line.substring(6).trim());
            this.delete(key);
            //System.out.println("key " + key + " has been deleted ");
        } else {
            DataObj obj = new DataObj();
            obj.setId(Long.valueOf(line.trim()));
            try {
                this.put(obj);
            } catch (Exception e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
                // result = "HEAVY_LOAD_NODE_FOUND";
            }
        }
        return result;
    }

    private void delete(Long key) {
        // TODO Auto-generated method stub
        BigInteger hash = hashFunction.generateSHA1(key.toString());
        if (!circle.containsKey(hash)) {
            SortedMap<BigInteger, VirtualNode> tailMap = circle.tailMap(hash);
            hash = tailMap.isEmpty() ? circle.firstKey() : tailMap.firstKey();
        }
        VirtualNode node = circle.get(hash);
        node.remove(key);
    }

    public StatisticParameters statistic() {
        EvaluatedParameters evaParams = evaluate();
        Evaluator evaluator = new Evaluator(evaParams, false);
        StatisticParameters res = new StatisticParameters();
        //System.out.println("total load : " + evaluator.getNetworkLoad());
//		System.out.println("% load : " + evaluator.getNetworkLoad()/Math.pow(2, ks) * 100);
        //System.out.println(" mean " + evaluator.getMean());
        double _std = evaluator.getStandardDeviation() / evaluator.getMean() * 100;
        res.setStandardDeviation(_std);
        res.setTotalLoad(evaluator.getNetworkLoad());
//		System.out.println(" std % " + _std);
//		System.out.println(" node max load " +  evaluator.getNodeMaxLoad().getID() + " with load: " + evaluator.getNodeMaxLoad().getLoad());
//		System.out.println(" node min load " +  evaluator.getNodeMinLoad().getID() + " with load: " + evaluator.getNodeMinLoad().getLoad());
        return res;

    }

    private void getbackData(VirtualNode target, VirtualNode source) throws Exception {
//        throw new UnsupportedOperationException("Not yet implemented");
        List<Long> keys = new ArrayList<Long>(target.listKeys());
        int count = 0;
        //two cases !very important
        if (circle.firstKey().longValue() == source.getID()) {
            System.out.println("target is first key");
            //target ID >= 0
            for (Long k : keys) {
                BigInteger hash = hashFunction.generateSHA1(k.toString());
                if (hash.longValue() > circle.lastKey().longValue()) {
                    source.put(k, target.get(k));
                    target.remove(k);
                    count++;
                }
            }
        } else {
            for (Long k : keys) {
                BigInteger hash = hashFunction.generateSHA1(k.toString());
                if (hash.longValue() <= source.getID()) {
                    source.put(k, target.get(k));
                    target.remove(k);
                    count++;
                }
            }
        }
        System.out.println("key " + source.getID() + "get back from " + target.getID() + " count " + count);
    }

    public int getNetworkStatus() {
        return networkStatus;
    }

    public void setNetworkStatus(int networkStatus) {
        this.networkStatus = networkStatus;
    }

    public void loadData(String text, long identityIdx) {
        try {
            for (long i = identityIdx; i < identityIdx + Integer.parseInt(text); i++) {
                DataObj obj = new DataObj();
                obj.setId(i);
                this.put(obj);
            }
        } catch (Exception e) {
        }
        setChanged();
        notifyObservers();
    }

    private void shareData(VirtualNode successorNode, VirtualNode vNode) throws Exception {
        long count = 0;
        for (Long storedKey : successorNode.listKeys()) {
            BigInteger hash = hashFunction.generateSHA1(storedKey.toString());
            if (circle.firstKey().longValue() == vNode.getID()) {
                if (hash.longValue() > circle.lastKey().longValue()) {
                    //transfer to new node
                    vNode.put(storedKey, successorNode.get(storedKey));
                    successorNode.remove(storedKey);
                    count++;
                }
            } else {
                if (hash.longValue() <= vNode.getID()) {
                    //transfer to new node
                    vNode.put(storedKey, successorNode.get(storedKey));
                    successorNode.remove(storedKey);
                    count++;
                }
            }
        }
        System.out.println(" count shared " + count);
    }
}
